package com.kygeng.datamonitor.listener;

import com.alibaba.fastjson.JSONObject;
import com.kygeng.common.config.KafkaConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.context.CanalContext;
import top.javatool.canal.client.handler.EntryHandler;
import top.javatool.canal.client.model.CanalModel;

import java.util.Map;

@CanalTable(value = "all")
@Component
public class DefaultEntryHandler implements EntryHandler<Map<String, String>> {

    private static Logger logger = LoggerFactory.getLogger(DefaultEntryHandler.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void insert(Map<String, String> map) {

        logger.info("insert message  {}", map);

        CanalModel canal = CanalContext.getModel();
        String database = canal.getDatabase();
        String table = canal.getTable();
        String companyCode = map.get("company_code");

        JSONObject message = new JSONObject();
        message.put("type", "insert");
        message.put("database", database);
        message.put("table", table);
        message.put("companyCode", companyCode);
        //message.put("data", map);

        kafkaTemplate.send(KafkaConfig.TOPIC_COMPANY, message.toJSONString());
    }


    @Override
    public void update(Map<String, String> before, Map<String, String> after) {

        logger.info("update before {} ", before);
        logger.info("update after {}", after);

        CanalModel canal = CanalContext.getModel();
        String database = canal.getDatabase();
        String table = canal.getTable();

        JSONObject message = new JSONObject();
        message.put("type", "insert");
        message.put("database", database);
        message.put("table", table);

        String companyCodeBefore = before.get("company_code");
        String companyCodeAfter = after.get("company_code");
        //如果主键发生了变化,则改为：发送一条删除消息，发送一条插入消息
        if (!StringUtils.isEmpty(companyCodeBefore) && !companyCodeBefore.equals(companyCodeAfter)) {
            //删除消息
            message.put("type", "delete");
            message.put("companyCode", companyCodeBefore);
            kafkaTemplate.send(KafkaConfig.TOPIC_COMPANY, message.toJSONString());

            //插入消息
            message.put("type", "insert");
            message.put("companyCode", companyCodeAfter);
            kafkaTemplate.send(KafkaConfig.TOPIC_COMPANY, message.toJSONString());
        }
        //如果主键没有发生变化，则：发送一条更新消息
        else {
            //插入消息
            message.put("type", "update");
            message.put("companyCode", companyCodeAfter);
            kafkaTemplate.send(KafkaConfig.TOPIC_COMPANY, message.toJSONString());
        }
    }

    @Override
    public void delete(Map<String, String> map) {

        logger.info("delete  {}", map);

        CanalModel canal = CanalContext.getModel();
        String database = canal.getDatabase();
        String table = canal.getTable();

        JSONObject message = new JSONObject();
        message.put("type", "insert");
        message.put("database", database);
        message.put("table", table);

        String companyCode = map.get("company_code");

        //删除消息
        message.put("type", "delete");
        message.put("companyCode", companyCode);
        kafkaTemplate.send(KafkaConfig.TOPIC_COMPANY, message.toJSONString());
    }
}